package mq

import (
	"fmt"
	"github.com/streadway/amqp"
	"go.uber.org/zap"
	"mall-portalv2/global"
	"sync"
	"time"
)

// 定义生产者接口
//type Producer interface {
//	MsgContent() string
//}

type Exchange struct {
	Name       string
	Kind       string
	Durable    bool
	AutoDelete bool
	Internal   bool
	NoWait     bool
	Arg        amqp.Table
}

type Queue struct {
	Name       string
	Durable    bool
	AutoDelete bool
	Internal   bool
	NoWait     bool
	Arg        amqp.Table
	Queue      amqp.Queue
}

type Producer struct {
	// 1 生产者 2 消费者
	Type     int `json:"type"`
	Exchange Exchange
	Queue    Queue
}

// 定义生产者接口
type DeadProducer struct {
	Exchange Exchange
	Queue    Queue
}

// 定义队列交换机对象
type RabbitMQConf struct {
	Dns string //链接地址
}

type RabbitMQ struct {
	Connection       *amqp.Connection
	Channel          *amqp.Channel
	Dns              string
	ConnectList      []Producer
	ProducerList     map[string]Producer
	DeadProducerList map[string]Producer
	Mutex            sync.RWMutex
	Wg               sync.WaitGroup
}

// 创建一个新的操作对象
func NewRabbitMQ(q *RabbitMQConf, connectClients []Producer) *RabbitMQ {
	conn, err := amqp.Dial(q.Dns)
	if err != nil {
		global.Log.Panic("Failed to connect to RabbitMQ", zap.Error(err))
	}
	ch, err := conn.Channel()
	if err != nil {
		global.Log.Panic("Failed to connect to RabbitMQ", zap.Error(err))
	}

	mq := &RabbitMQ{
		Dns:              q.Dns,
		Connection:       conn,
		Channel:          ch,
		ConnectList:      connectClients,
		ProducerList:     make(map[string]Producer),
		DeadProducerList: make(map[string]Producer),
	}
	go mq.Start()
	return mq
}

// 关闭RabbitMQ连接
func (r *RabbitMQ) Close() {
	// 先关闭管道,再关闭链接
	err := r.Channel.Close()
	if err != nil {
		fmt.Printf("MQ管道关闭失败:%s \n", err)
	}
	err = r.Connection.Close()
	if err != nil {
		fmt.Printf("MQ链接关闭失败:%s \n", err)
	}
}

// 启动RabbitMQ客户端,并初始化
func (r *RabbitMQ) Start() (err error) {
	//r.Connection = conn
	//r.Channel = ch

	for _, producer := range r.ConnectList {
		if producer.Type == 1 {
			// 开启监听生产者发送任务
			if err = r.listenProducer(producer); err != nil {
				global.Log.Panic("listenProducer error", zap.Error(err))
				return err
			}
		} else {
			// 开启监听接收者接收任务
			r.Wg.Add(1)
			go func() {
				if err = r.listenReceiver(producer); err != nil {
					global.Log.Panic("listenReceiver error", zap.Error(err))
				}
			}()
		}
	}
	r.Wg.Wait()
	time.Sleep(time.Microsecond * 1)
	return err
}

func (r *RabbitMQ) listenProducer(producer Producer) error {
	// 初始化对应的交换机
	if err := r.Channel.ExchangeDeclare(producer.Exchange.Name, producer.Exchange.Kind, producer.Exchange.Durable, producer.Exchange.AutoDelete, producer.Exchange.Internal, producer.Exchange.NoWait, producer.Exchange.Arg); err != nil {
		return err
	}

	//amqp.Table{
	//		"x-message-ttl":             5000,           // 消息过期时间,毫秒
	//		"x-dead-letter-exchange":    "zxz.ttl.test", // 指定死信交换机
	//		"x-dead-letter-routing-key": "",
	//	}
	// 初始化对应的队列
	q, err := r.Channel.QueueDeclare(producer.Queue.Name, producer.Queue.Durable, producer.Queue.AutoDelete, producer.Queue.Internal, producer.Queue.NoWait, producer.Queue.Arg)
	if err != nil {
		return err
	}
	//队列bind交换机
	err = r.Channel.QueueBind(q.Name, "", producer.Exchange.Name, false, nil)

	producer.Queue.Queue = q

	temp := fmt.Sprintf("%s", producer.Queue.Name)

	global.Log.Info("producerList join", zap.Any("data", temp))

	r.ProducerList[temp] = producer

	return nil
}

func (r *RabbitMQ) listenReceiver(producer Producer) error {
	// 初始化对应的交换机
	if err := r.Channel.ExchangeDeclare(producer.Exchange.Name, producer.Exchange.Kind, producer.Exchange.Durable, producer.Exchange.AutoDelete, producer.Exchange.Internal, producer.Exchange.NoWait, producer.Exchange.Arg); err != nil {
		return err
	}

	//amqp.Table{
	//		"x-message-ttl":             5000,           // 消息过期时间,毫秒
	//		"x-dead-letter-exchange":    "zxz.ttl.test", // 指定死信交换机
	//		"x-dead-letter-routing-key": "",
	//	}
	// 初始化对应的队列
	q, err := r.Channel.QueueDeclare(producer.Queue.Name, producer.Queue.Durable, producer.Queue.AutoDelete, producer.Queue.Internal, producer.Queue.NoWait, producer.Queue.Arg)
	if err != nil {
		return err
	}
	//队列bind交换机
	err = r.Channel.QueueBind(q.Name, "", producer.Exchange.Name, false, nil)

	producer.Queue.Queue = q

	temp := fmt.Sprintf("%s", producer.Queue.Name)

	global.Log.Info("deadProducerList join", zap.Any("data", temp))

	r.DeadProducerList[temp] = producer

	return nil
}
